package com.jeffse.light.stream.core.utils;

import org.apache.pulsar.client.api.*;

import java.util.List;
import java.util.concurrent.TimeUnit;


public class PulsarUtils {
    private static PulsarUtils INSTANCE = new PulsarUtils();
    private PulsarClient pulsarClient = null;
    private Consumer consumer = null;
    private Producer producer = null;
    private final static int DEFAULT_PENDING_MESSAGES = 5_000;
    private final static int DEFAULT_SEND_TIMEOUT = 10;

    public static PulsarUtils getINSTANCE(){
        return INSTANCE;
    }

    public PulsarClient getClient(String serviceUrl) throws PulsarClientException {
        if(pulsarClient == null){
            pulsarClient =  PulsarClient.builder()
                    .serviceUrl(serviceUrl)
                    .listenerThreads(10)
                    .build();
        }

        return pulsarClient;
    }

    public Consumer getConsumer(String topic, String group, MessageListener listener) throws PulsarClientException {
        if(consumer == null){
            consumer = pulsarClient.newConsumer().topic(topic)
                    .subscriptionName(group)
                    .ackTimeout(10, TimeUnit.SECONDS)
                    .subscriptionType(SubscriptionType.Key_Shared)
                    .messageListener(listener)
                    .subscribe();
        }

        return consumer;
    }

    public Consumer getConsumer(List<String> topics, String group, MessageListener listener) throws PulsarClientException {
        if(consumer == null){
            consumer = pulsarClient.newConsumer().topics(topics)
                    .subscriptionName(group)
                    .ackTimeout(10, TimeUnit.SECONDS)
                    .subscriptionType(SubscriptionType.Key_Shared)
                    .messageListener(listener)
                    .subscribe();
        }

        return consumer;
    }

    public Producer getProducer(String topic) throws PulsarClientException {
        if(producer == null){
            producer = pulsarClient.newProducer(Schema.STRING)
                    .maxPendingMessages(DEFAULT_PENDING_MESSAGES)
                    .blockIfQueueFull(true)
                    .enableBatching(true)
                    .blockIfQueueFull(true)
                    .sendTimeout(DEFAULT_SEND_TIMEOUT, TimeUnit.SECONDS)
                    .topic("test")
                    .create();
        }
        return producer;
    }

    public void removeConsumer(){
        try {
            this.consumer.close();
        } catch (PulsarClientException e) {
            e.printStackTrace();
        }
    }
}
